package com.apps.sdses.flink141.demo;

import com.apps.sdses.flink141.udfs.UDFGetFieldFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/9/13 10:46
 */
public class UnionDemo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //注册UDF
        tableEnv.createTemporaryFunction("getField", UDFGetFieldFunction.class);

        //Source
        tableEnv.executeSql("CREATE TABLE sourceTable1 (\n" +
                "    id       bigint,\n" +     //设备编号
                "    name      string,\n" +    //设备名称
                "    in_pres   double,\n" +    //进站压力
                "    out_pres   double,\n" +    //出站压力
                "    in_temp   double,\n" +    //进站温度
                "    out_temp   double,\n" +    //出战温度
                "    real_flow   double,\n" +    //瞬时流量
                "    read_time   timestamp(3)\n" +       //采集时间
                ") WITH (\n" +
                "  'connector' = 'datagen', \n" +
//                "  'number-of-rows' = '5', \n" +
                "  'rows-per-second' = '1', \n" +
                "  'fields.name.length' = '5', \n" +
                "  'fields.id.min' = '101', \n" +
                "  'fields.id.max' = '102', \n" +
                "  'fields.in_pres.min' = '10', \n" +
                "  'fields.in_pres.max' = '100', \n" +
                "  'fields.out_pres.min' = '10', \n" +
                "  'fields.out_pres.max' = '100', \n" +
                "  'fields.in_temp.min' = '0', \n" +
                "  'fields.in_temp.max' = '200', \n" +
                "  'fields.out_temp.min' = '0', \n" +
                "  'fields.out_temp.max' = '200', \n" +
                "  'fields.real_flow.min' = '10', \n" +
                "  'fields.real_flow.max' = '200'\n" +
                ")");


        tableEnv.executeSql("CREATE TABLE sourceTable2 (\n" +
                "    id       bigint,\n" +     //设备编号
                "    name      string,\n" +    //设备名称
                "    in_pres   double,\n" +    //进站压力
                "    out_pres   double,\n" +    //出站压力
                "    in_temp   double,\n" +    //进站温度
                "    out_temp   double,\n" +    //出战温度
                "    real_flow   double,\n" +    //瞬时流量
                "    read_time   timestamp(3)\n" +       //采集时间
                ") WITH (\n" +
                "  'connector' = 'datagen', \n" +
//                "  'number-of-rows' = '5', \n" +
                "  'rows-per-second' = '1', \n" +
                "  'fields.name.length' = '5', \n" +
                "  'fields.id.min' = '103', \n" +
                "  'fields.id.max' = '104', \n" +
                "  'fields.in_pres.min' = '10', \n" +
                "  'fields.in_pres.max' = '100', \n" +
                "  'fields.out_pres.min' = '10', \n" +
                "  'fields.out_pres.max' = '100', \n" +
                "  'fields.in_temp.min' = '0', \n" +
                "  'fields.in_temp.max' = '200', \n" +
                "  'fields.out_temp.min' = '0', \n" +
                "  'fields.out_temp.max' = '200', \n" +
                "  'fields.real_flow.min' = '10', \n" +
                "  'fields.real_flow.max' = '200'\n" +
                ")");

        tableEnv.executeSql("CREATE TEMPORARY VIEW unionTable(id, name, in_pres, out_pres, in_temp, out_temp, real_flow, read_time) as " +
                " select * from sourceTable1 " +
                " union all " +
                " select * from sourceTable2");


        tableEnv.executeSql("select * from unionTable where getField(row)");


//        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
//                "LIKE sourceTable1 (EXCLUDING ALL)");

        //Insert
        tableEnv.executeSql("insert into sinkTable select * from unionTable");
    }
}
